/*

 * Licensed to the Apache Software Foundation (ASF) under one

 * or more contributor license agreements.  See the NOTICE file

 * distributed with this work for additional information

 * regarding copyright ownership.  The ASF licenses this file

 * to you under the Apache License, Version 2.0 (the

 * "License"); you may not use this file except in compliance

 * with the License.  You may obtain a copy of the License at

 *

 *     http://www.apache.org/licenses/LICENSE-2.0

 *

 * Unless required by applicable law or agreed to in writing, software

 * distributed under the License is distributed on an "AS IS" BASIS,

 * WITHOUT WARRANTIES OR CONDITIONS OF ANY KIND, either express or implied.

 * See the License for the specific language governing permissions and

 * limitations under the License.

 */

package com.bff.gaia.unified.sdk.extensions.sql.impl;



import com.bff.gaia.unified.sdk.extensions.sql.meta.Table;

import com.bff.gaia.unified.sdk.extensions.sql.meta.provider.TableProvider;

import org.apache.calcite.linq4j.tree.Expression;

import org.apache.calcite.rel.type.RelProtoDataType;

import org.apache.calcite.schema.*;



import java.util.*;



/** Adapter from {@link TableProvider} to {@link Schema}. */

public class UnifiedCalciteSchema implements Schema {

  private JdbcConnection connection;

  private TableProvider tableProvider;

  private Map<String, UnifiedCalciteSchema> subSchemas;



  UnifiedCalciteSchema(JdbcConnection jdbcConnection, TableProvider tableProvider) {

    this.connection = jdbcConnection;

    this.tableProvider = tableProvider;

    this.subSchemas = new HashMap<>();

  }



  public TableProvider getTableProvider() {

    return tableProvider;

  }



  public Map<String, String> getPipelineOptions() {

    return connection.getPipelineOptionsMap();

  }



  public void setPipelineOption(String key, String value) {

    Map<String, String> options = new HashMap<>(connection.getPipelineOptionsMap());

    options.put(key, value);

    connection.setPipelineOptionsMap(options);

  }



  public void removePipelineOption(String key) {

    Map<String, String> options = new HashMap<>(connection.getPipelineOptionsMap());

    options.remove(key);

    connection.setPipelineOptionsMap(options);

  }



  public void removeAllPipelineOptions() {

    connection.setPipelineOptionsMap(Collections.emptyMap());

  }



  @Override

  public boolean isMutable() {

    return true;

  }



  @Override

  public Schema snapshot(SchemaVersion version) {

    return this;

  }



  @Override

  public Expression getExpression(SchemaPlus parentSchema, String name) {

    return Schemas.subSchemaExpression(parentSchema, name, getClass());

  }



  @Override

  public Set<String> getTableNames() {

    return tableProvider.getTables().keySet();

  }



  @Override

  public RelProtoDataType getType(String name) {

    return null;

  }



  @Override

  public Set<String> getTypeNames() {

    return Collections.emptySet();

  }



  @Override

  public org.apache.calcite.schema.Table getTable(String name) {

    Table table = tableProvider.getTable(name);

    if (table == null) {

      return null;

    }

    return new UnifiedCalciteTable(tableProvider.buildUnifiedSqlTable(table), getPipelineOptions());

  }



  @Override

  public Set<String> getFunctionNames() {

    return Collections.emptySet();

  }



  @Override

  public Collection<Function> getFunctions(String name) {

    return Collections.emptySet();

  }



  @Override

  public Set<String> getSubSchemaNames() {

    return tableProvider.getSubProviders();

  }



  @Override

  public Schema getSubSchema(String name) {

    if (!subSchemas.containsKey(name)) {

      TableProvider subProvider = tableProvider.getSubProvider(name);

      UnifiedCalciteSchema subSchema =

          subProvider == null ? null : new UnifiedCalciteSchema(connection, subProvider);

      subSchemas.put(name, subSchema);

    }

    return subSchemas.get(name);

  }

}